package qgs.zisu.flink.flink;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.lang.reflect.Parameter;

/**
 * @Author: 蔡翔
 * @Date: 2019/9/2 18:44
 * @Version 1.0
 *
 * 通过socket模拟产生单词数据（滑动窗口计算）
 * flink对数据进行统计计算
 *
 * 需要实现每隔一秒对最近两秒的数据进行汇总计算
 *
 * 执行虚拟机条件： 需要在虚拟机上  执行 nc -l 9000
 */
public class SocketWindowWordCountJava {
    public static void main(String[] args) throws Exception{
        //获取需要的端口号
        int port = 9000;
        try {
            ParameterTool parameterTool = ParameterTool.fromArgs(args);
            port = parameterTool.getInt("port"); //通过控制台传参数
        }catch (Exception e){
            System.err.println("No Port set,use default 9000");
            port = 9000;
        }
        //获取flink的运行环境
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        String hostname = "192.168.2.125";
        String delimiter = "\n";

        // 连接socket 获取输入的数据。
        DataStreamSource<String> text = executionEnvironment.socketTextStream(hostname, port, delimiter);

        // a b c
        // a 1
        // b 1
        // c 1
        SingleOutputStreamOperator<WordWithCount> windowCounts = text.flatMap(new FlatMapFunction<String, WordWithCount>() {  // <String, Object> 代表传进来的是string  返回的是Object

            @Override
            public void flatMap(String value, Collector<WordWithCount> collector) throws Exception {
                String[] s = value.split("\\s");
                for (String word : s) {
                    collector.collect(new WordWithCount(word, 1L));
                }
            }
        }).keyBy("word")  //把收集过来的collect依据word来分组。
                .timeWindow(Time.seconds(2), Time.seconds(1))  //指定时间窗口大小为2秒，指定时间间隔为1s
                .sum("count"); //这里使用sum或者reduce都可以。意思是按照分好的组进行count相加。
                /*.reduce(new ReduceFunction<WordWithCount>() {
                    @Override
                    public WordWithCount reduce(WordWithCount a, WordWithCount b) throws Exception {
                        return new WordWithCount(a.word, a.count + b.count);
                    }
                })*/;
        //把数据打印到控制台，并且设置并行度为1
        windowCounts.print().setParallelism(1);
        //这一行代码一定要实现，否则程序不执行
        executionEnvironment.execute("Socket window count");

    }
    public static class WordWithCount{
        public String word;
        public long count;
        public WordWithCount(){

        }
        public WordWithCount(String word,long count){
            this.word = word;
            this.count = count;
        }

        @Override
        public String toString() {
            return "word:"+this.word+",count:"+this.count;
        }
    }
}
